package com.zzx.graduation.controller;

import com.alibaba.fastjson.JSON;
import com.zzx.graduation.entity.KafkaNews;
import com.zzx.graduation.entity.News;
import com.zzx.graduation.utils.KafkaBatchConsumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

@Component
@ServerEndpoint("/webSocket")
@Slf4j
public class WebSocket {

    private Session session;
    //如果消息发往群聊，则消息要发到kafka,主题tpoic为：group--targetid(群id)
    //发往个人主题邮箱topic： mail--id （代表这个id的人的收件箱，存放收到的消息）
    private Integer id;
    KafkaBatchConsumer kafkaBatchConsumer;
    private Thread thread = new Thread(new Runnable() {
        @Override
        public void run() {
            //  测试消费
            kafkaBatchConsumer = new KafkaBatchConsumer("mail--" + id, id + "") {
                @Override
                public void hanldeRecord(ConsumerRecord<String, String> record) {
                    String str = record.value();
                    System.out.println("推送消息消费--->" + str );
                    //消息推送
                    try {
                     // session.getBasicRemote().sendObject(news);
                        session.getBasicRemote().sendText(str);
                        System.out.println("消息推送成功");
                    } catch (Exception e) {
                        System.out.println("消息推送失败");
                        e.printStackTrace();
                    }
                }
            };
            kafkaBatchConsumer.receivebatch();
        }
    });

    private static CopyOnWriteArraySet<WebSocket> webSocketSet = new CopyOnWriteArraySet<>();
     private static ConcurrentHashMap<Integer,WebSocket> webSocketMap = new ConcurrentHashMap<>();


    @OnOpen
    public void onOpen(Session session) {
        this.session = session;

        String[] split = session.getQueryString().split("=");
        System.out.println("传入的参数为：" + split[1]);

        if (split[0].equals("id") && !webSocketMap.containsKey(Integer.valueOf(split[1]))) {
            this.id = Integer.valueOf(split[1]);
            webSocketSet.add(this);
            webSocketMap.put(Integer.valueOf(split[1]), this);
            log.info("【websocket消息】有新的连接, 总数:{}", webSocketSet.size());
            //确定登录状态，开启消费者，开始接受消息，拉取kafka
            this.thread.start();
        }else {
            try {
                session.getBasicRemote().sendText("没有传入id或者 此id已经在线，正在进行处理");
            } catch (IOException e) {
                e.printStackTrace();
            }finally {
                onClose();

            }
        }

    }

    @OnClose
    public void onClose() {
        webSocketSet.remove(this);
        webSocketMap.remove(this.id);
        //关闭kafka消费者
        this.kafkaBatchConsumer.flag = false;
        log.info("【websocket消息】连接断开, 总数:{}", webSocketSet.size());
    }

    @OnMessage
    public void onMessage(String message) {
        log.info("【websocket消息】收到客户端发来的消息:{}", message);
    }

    public void sendMessage(String message) {
        for (WebSocket webSocket : webSocketSet) {
            log.info("【websocket消息】广播消息, message={}", message);
            try {
                webSocket.session.getBasicRemote().sendText(message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

}
